#include "keeper.h"

static const string MODULE = "KEEPER";

Keeper::Keeper()
{
	handler = NULL;
	connected = false;
	zoo_set_debug_level(ZOO_LOG_LEVEL_INFO);
}

Keeper::~Keeper()
{
	close();
}

int Keeper::open(const Config &conf)
{
	string zoo_addr = conf.zoo_host + ":" + std::to_string(conf.zoo_port);

	clientid_t cli_tid;
	memset(&cli_tid, 0, sizeof(cli_tid));

	handler = zookeeper_init(zoo_addr.c_str(), watcher, conf.timeout, &cli_tid, (void*)this, 0);
	if(!handler)
		return 1;

	config = conf;
	return 0;
}

void Keeper::close()
{
	if(handler)
	{
		zookeeper_close(handler);
		handler = NULL;
	}
}

int Keeper::checkin(const string &label, const string &node)
{
	if (!connected)
	    return 1;

	struct ACL acl[] = {{ZOO_PERM_ALL, ZOO_ANYONE_ID_UNSAFE}};
	struct ACL_vector acl_vec = {1, acl};

	string path = label + "/" + node;
	char buf[256];
	int ret = 0;

	ret = zoo_create(handler, path.c_str(), "", 0, &acl_vec, ZOO_EPHEMERAL, buf, sizeof(buf));
	if(ret != ZOK)
	{
		logger.log_error(MODULE, "fail to checkin %s: %d", path.c_str(), ret);
		return 2;
	}

	return 0;
}

int Keeper::resolve(const string &label, string &node)
{
	std::unique_lock<std::mutex> lock(cache_mutex);

	std::vector<string> &cache = cache_map[label];

	if (!cache.empty())
	{
		node = cache[rand()%cache.size()];
		logger.log_debug("resolve %s => %s", label.c_str(), node.c_str());
		return 0;
	}

	if (!connected)
	    return 1;

	String_vector children;
	int ret;

	ret = zoo_get_children(handler, label.c_str(), 1, &children);
	if (ret != ZOK)
	{
		logger.log_notice(MODULE, "fail to get sub of %s: %d", label.c_str(), ret);
		return 2;
	}

	if (children.count <= 0)
	{
		logger.log_notice(MODULE, "nothing in sub of %s", label.c_str());
		return 3;
	}

	cache.resize(children.count);

	for (int i=0; i<children.count; ++i)
		cache[i] = children.data[i];

	node = cache[rand()%cache.size()];
	logger.log_debug(MODULE, "resolve %s => %s", label.c_str(), node.c_str());
	return 0;
}

void Keeper::main_watcher(int type, int state, const char *path)
{
    logger.log_notice(MODULE, "watcher: type:%d, state:%d", type, state);

    if (type == ZOO_SESSION_EVENT)
    {
        if (state == ZOO_CONNECTED_STATE)
        {
            logger.log_notice(MODULE, "keeper to connected state");
            connected = true;
        }
        else if (state == ZOO_EXPIRED_SESSION_STATE)
        {
            logger.log_notice(MODULE, "keeper to expired state");
            connected = false;

            logger.log_notice(MODULE, "keeper to open again");
            close();
            open();
        }
        else
        {
            logger.log_notice(MODULE, "keeper to disconn state");
            connected = false;
        }

        return;
    }

    if (path)
    {
        logger.log_notice(MODULE, "watcher: path changed state [%s] ", path);
        std::unique_lock<std::mutex> lock(cache_mutex);
        cache_map[path].clear();
    }
}

void Keeper::watcher(zhandle_t *zh, int type, int state, const char *path, void *context)
{
    Keeper *keeper = (Keeper *)context;
    keeper->main_watcher(type, state, path);
}

